package cn.com.sparkle.paxos.net.protocolprocessor;

import org.apache.log4j.Logger;

import cn.com.sparkle.paxos.net.netlayer.PaxosSession;

public class DefaultProtocolProcessorChain<T> implements ProtocolProcessorChain<T> {
	private Logger logger = Logger.getLogger(DefaultProtocolProcessorChain.class);

	ProtocolProcessorExecutor<T> first = new ProtocolProcessorExecutor<T>(new ProtocolProcessor<T>() {
		@Override
		public boolean receive(T obj, PaxosSession session) {
			logger.warn("unprocessed message");
			return true;
		}

		@Override
		public void onConnect(PaxosSession session) {
		}

		@Override
		public void onDisConnect(PaxosSession session) {
		}
	}, null);

	@Override
	public void addFirst(ProtocolProcessor<T> protocolProcessor) {
		ProtocolProcessorExecutor<T> executor = new ProtocolProcessorExecutor<T>(protocolProcessor, this.first);
		this.first = executor;
	}

	@Override
	public boolean receive(T t, PaxosSession session) {
		try {
			return first.receive(t, session);
		} catch (RuntimeException e) {
			session.closeSession();
			logger.error("", e);
			throw e;
		}
	}

	@Override
	public void onConnect(PaxosSession session) {
		first.onConnect(session);
	}

	@Override
	public void onDisConnect(PaxosSession session) {
		first.onDisConnect(session);
	}

	private final static class ProtocolProcessorExecutor<T> implements ProtocolProcessor<T> {
		private ProtocolProcessorExecutor<T> next;
		private ProtocolProcessor<T> processor;

		public ProtocolProcessorExecutor(ProtocolProcessor<T> processor, ProtocolProcessorExecutor<T> next) {
			this.processor = processor;
			this.next = next;
		}

		@Override
		public boolean receive(T t, PaxosSession session) {
			if (!processor.receive(t, session)) {
				if (next != null) {
					next.receive(t, session);
				}
			}
			return true;
		}

		@Override
		public void onConnect(PaxosSession session) {
			processor.onConnect(session);
			if (next != null) {
				next.onConnect(session);
			}
		}

		@Override
		public void onDisConnect(PaxosSession session) {
			processor.onConnect(session);
			if (next != null) {
				next.onConnect(session);
			}
		}
	}

}
